package com.joven.core.utils.lockUtil;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.CountDownLatch;

/**
 * @author joven_wong
 * 2021/6/29 16:09
 * Curator提供了来实现分布式锁，用方法获取锁，以及用方法释放锁
 * zookeeper 可以根据有序节点+watch实现，实现思路，如：
 * 为每个线程生成一个 有序的临时节点，为确保有序性，在排序一次全部节点，获取全部节点，每个线程判断自己是否最小，
 * 如果是的话，获得锁，执行操作，操作完删除自身节点。
 * 如果不是第一个的节点则监听它的前一个节点，当它的前一个节点被删除时，则它会获得锁，以此类推。
 */
@Slf4j
@Service
public class ZkDistributedLock implements InitializingBean {
    private static final String ROOT_PATH_LOCK = "zklock";
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    @Autowired
    private CuratorFramework curatorFramework;

    /**
     * 获取锁
     */
    public void acquireLock(String path){
        String keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
        //临时节点
        while (true){
            try{
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        //完全开放的ACL，任何连接的客户端都可以操作该属性znode
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(keyPath);
                log.info("SUCCESS to acquire lock for path:{}",keyPath);
                break;
            }catch (Exception e){
                log.info("FAIL to acquire lock for path:{}",keyPath);
                try {
                    if(countDownLatch.getCount() <= 0){
                        countDownLatch = new CountDownLatch(1);
                    }
                    countDownLatch.await();
                }catch (InterruptedException e1){
                    e1.printStackTrace();
                }
            }
        }
    }

    /**
     * 释放锁
     */
    public boolean release(String path){
        try{
            String keyPath = "/" + ROOT_PATH_LOCK + "/" + path;
            if(curatorFramework.checkExists().forPath(keyPath) != null){
                curatorFramework.delete().forPath(keyPath);
            }
        }catch (Exception e){
            log.info("FAIL to release lock ");
            return false;
        }
        return true;
    }

    private void addWatcher(String path) throws Exception{
        String keyPath;
        if(ROOT_PATH_LOCK.equals(path)){
            keyPath = "/" + path;
        } else{
            keyPath = "/" + ROOT_PATH_LOCK + "/" + path;;
        }
        final PathChildrenCache cache = new PathChildrenCache(curatorFramework, keyPath, false);
        //异步初始化。初始化后会触发事件
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        //curatorFramework == client;PathChildrenCacheEvent == event
        cache.getListenable().addListener((client, event) -> {
            if(PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(event.getType())){
                String oldPath = event.getData().getPath();
                log.info("上一个节点" + oldPath + "已经被断开开");
                if(oldPath.contains(path)){
                    //释放计数器，让当前的请求获取锁
                    countDownLatch.countDown();
                }
            }
        });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        //初始加载，持久化节点
        curatorFramework = curatorFramework.usingNamespace("lock-namespace");
        String path = "/" + ROOT_PATH_LOCK;
        try {
            if(curatorFramework.checkExists().forPath(path) == null){
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(path);
            }
            addWatcher(ROOT_PATH_LOCK);
        }catch (Exception e){
            log.error("connect zookeeper fail");
        }
    }
}
